#include "FilterPipeSystem.h"
#include "Common.h"
using namespace AutomatedSurfaceReconstruction;

CFilterPipeSystem::CFilterPipeSystem()
{
	hiveEventLogger::hiveOutputEvent("Event description: Init FilterPipeSystem...", "SurfaceReconstruction", "FilterPipeSystem.cpp", 7, hiveEventLogger::EEventLevel::EVENT_LEVEL_INFO);
	__initXML();
	__initFilter();
	__initPipe();
	__initExternalBuffer();
	m_ReconstructSystem.launch();
}

CFilterPipeSystem::~CFilterPipeSystem()
{
	m_ReconstructSystem.shutdown();
}

bool CFilterPipeSystem::pushData2FilePool(std::pair<std::pair<std::string, CGALPointSetType>, CGALPointSetType>& vPointSet2DXFRoad)
{
	m_FilePool.push_back(vPointSet2DXFRoad);
	return true;
}

std::string CFilterPipeSystem::setMaxUsedThread()
{
	HANDLE CurrentProcessHandle = GetCurrentProcess();
	PROCESS_MEMORY_COUNTERS MemoryCounter;

	GetProcessMemoryInfo(CurrentProcessHandle, &MemoryCounter, sizeof(MemoryCounter));
	double BeforePeakMemory = MemoryCounter.PeakWorkingSetSize / 1024 / 1024;

	m_ReconstructSystem.push(m_FilePool.front(), m_InputBufferHandle.at(0).value());
	m_FilePool.pop_front();
	std::string TempPath;
	OutputSemaphore.acquire();
	while (1) { if (m_ReconstructSystem.try_pop(m_OutputBufferHandle.at(0).value(), TempPath) && !TempPath.empty())break; }

	GetProcessMemoryInfo(CurrentProcessHandle, &MemoryCounter, sizeof(MemoryCounter));
	double AfterPeakMemory = MemoryCounter.PeakWorkingSetSize / 1024 / 1024;
	double PeakMemory = AfterPeakMemory - BeforePeakMemory;
	double FreeMemory = getSystemTotalMemoryInMB();

	m_MaxUsedThread = FreeMemory / PeakMemory;
	if (m_MaxUsedThread > m_ProcessFileNum)m_MaxUsedThread = m_ProcessFileNum;
	else if (m_MaxUsedThread == 0)m_MaxUsedThread = 1;

	return TempPath;
}

bool CFilterPipeSystem::organizeExternalInputBuffer()
{
	int FileNum = m_FilePool.size();
	for (int i = 0; i < FileNum; i++)
	{
		m_ReconstructSystem.push(m_FilePool.front(), m_InputBufferHandle.at(i % m_MaxUsedThread).value());
		m_FilePool.pop_front();
	}
	return true;
}

std::string CFilterPipeSystem::getFinishedThreadOutputPath()
{
	std::string OutputFilePath;
	OutputSemaphore.acquire();
	bool GetOutputPathSuccess = false;
	while (!GetOutputPathSuccess)
	{
		for (int i = 0; i < m_MaxUsedThread; i++)
		{
			if (m_ReconstructSystem.try_pop(m_OutputBufferHandle.at(i).value(), OutputFilePath) && !OutputFilePath.empty()) { GetOutputPathSuccess = true; }
		}
	}
	return OutputFilePath;
}

void CFilterPipeSystem::__initXML()
{
	hiveConfig::hiveParseConfig(m_FilterPipeConfigFileName, hiveConfig::EConfigType::XML, CFilterPipeConfig::getInstance());
	bool InitSuccess = CFilterPipeConfig::getInstance()->init();

	m_FilterSet = CFilterPipeConfig::getInstance()->getFilterSet();
	m_PipeSet = CFilterPipeConfig::getInstance()->getPipeSet();
}

void CFilterPipeSystem::__initFilter()
{
	for (int i = 0; i < m_FilterSet.size(); i++)
	{
		SFilter TempFilter(m_FilterSet.at(i));
		hiveSystem::HKernel TempHandle = m_ReconstructSystem.addFilter(TempFilter._Sig, TempFilter._WorkThreadID);
		m_ID2Handle.insert(std::make_pair(TempFilter._FilterID, TempHandle));
		for (int k = 0; k < TempFilter._Connections.size(); k++)
		{
			Eigen::Vector2i Connection = TempFilter._Connections.at(k);
			m_ReconstructSystem.connectSingleStreamFilterInputBuffer2OutputBuffer(TempHandle, Connection.x(), Connection.y());
		}
	}
}

void CFilterPipeSystem::__initPipe()
{
	for (int i = 0; i < m_PipeSet.size(); i++)
	{
		SPipe TempPipe(m_PipeSet.at(i));
		hiveSystem::HKernel SrcFilterHandle = m_ID2Handle.at(TempPipe._SrcFilter), DstFilterHandle = m_ID2Handle.at(TempPipe._DstFilter);
		m_ReconstructSystem.connectFilter2Filter(SrcFilterHandle, TempPipe._SrcFilterBufferID, DstFilterHandle, TempPipe._DstFilterBufferID, TempPipe._Sig, TempPipe._WorkThreadID);
	}
}

void CFilterPipeSystem::__initExternalBuffer()
{
	for (int i = 0; i < m_ProcessFileNum; i++)
	{
		hiveSystem::CConcurrentQueue<std::pair<std::pair<std::string, CGALPointSetType>, CGALPointSetType>>* pExternalInputBuffer = new hiveSystem::CConcurrentQueue<std::pair<std::pair<std::string, CGALPointSetType>, CGALPointSetType>>();
		hiveSystem::CConcurrentQueue<std::string>* pExternalOutputBuffer = new hiveSystem::CConcurrentQueue<std::string>();
		m_pExternalInputBuffer.emplace_back(pExternalInputBuffer);
		m_pExternalOutputBuffer.emplace_back(pExternalOutputBuffer);
		std::optional<std::uint8_t> InputHandle = m_ReconstructSystem.attachExternalInputBuffer2Filter(pExternalInputBuffer, m_ID2Handle.at(m_SourceFilterIndex), i);
		bool InitInputHandleSuccess = InputHandle.has_value();
		m_InputBufferHandle.emplace_back(InputHandle);

		std::optional<std::uint8_t> OutputHandle = m_ReconstructSystem.attachExternalOutputBuffer2Filter(pExternalOutputBuffer, m_ID2Handle.at(m_DstFilterIndex), i);
		bool InitOutputHandleSuccess = OutputHandle.has_value();
		m_OutputBufferHandle.emplace_back(OutputHandle);
	}
}
